package io.kimmking.dubbo.common.dubboLB;

import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.LoadBalance;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;

@Slf4j
@Component
public class YlLoadBalance implements LoadBalance{
    //自定义的loadbalance类里的变量和方法需要保证线程安全性
    private Map<String,String> sessionIdAddress = new ConcurrentHashMap<String,String>(10);
    private LongAdder index = new LongAdder();

    //方法一：尝试使用定时器来实现，待实现
//    volatile private static Map<String,LongAdder> invokerCountSum = new ConcurrentHashMap<String,LongAdder>(2);
//
//    volatile private static Map<String,Long> oldInvokerCountSum = new ConcurrentHashMap<String,Long>(2);
//
//    volatile private static Map<String,Boolean> isAllowInvoke = new ConcurrentHashMap<String,Boolean>(2);
//
//    private static ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
//
//    //实现"如果一分钟内consumer调用某个服务提供者超过10次，则拒绝提供服务直到下一分钟"的需求,待完成
//    static {
//        executorService.scheduleAtFixedRate(new Runnable() {
//                                                @Override
//                                                public void run() {
//                                                    invokerCountSum.forEach((k,v) -> );
//
//                                                }
//                                            },
//                10,
//                60,
//                TimeUnit.SECONDS
//        );
//    }
    //参照dubbo里的限流的实现方法来实现
    private final ConcurrentMap<String, myStatItem> stats = new ConcurrentHashMap<String, myStatItem>();

    @Override
    public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
        Invoker result = null;
        //约定方法的第一个参数就是sessionId
        String sessionId = (invocation.getArguments()[0]).toString();
        Arrays.stream(invocation.getArguments()).parallel().forEach(i -> log.info("invocation's arguments:" + i.toString()));
        if(!sessionIdAddress.containsKey(sessionId)){
            index.increment();
            result = invokers.get(index.intValue()%invokers.size());
            sessionIdAddress.put(sessionId,result.getUrl().getAddress());
        }else{
            String destAddress = sessionIdAddress.get(sessionId);
            for (Invoker<T> invoker : invokers) {
                if(invoker.getUrl().getAddress().equals(destAddress)){
                    result = invoker;
                    break;
                }
            }
        }
        log.info("sesisonId: " + sessionId + " ,method: " + invocation.getMethodName() + " ,select " + result.getUrl().getAddress() + " broker");

//        invokerCountSum.computeIfAbsent(result.getUrl().getAddress(), k -> new LongAdder()).increment();

        //进行限速检查判断，如果超过设置的10Tpmin，则抛出异常，拒绝本次调用
        if(!isAllowable(result)){
            throw new RpcException(
                    "Failed to invoke service " +
                            result.getInterface().getName() +
                            "." +
                            invocation.getMethodName() +
                            " because exceed max service tps (value=10).");
        }

        return result;
    }

    private boolean isAllowable(Invoker invoker) {
        //获取tps这个参数设置的大小,这里设置固定值
        int rate = 10;
        //获取tps.interval这个参数设置的大小，这里设置为固定值60秒
        long interval = 60*1000;
//        String serviceKey = url.getServiceKey();
        String serviceKey = invoker.getUrl().getAddress();
        if (rate > 0) {
            myStatItem statItem = stats.get(serviceKey);
            if (statItem == null) {
                stats.putIfAbsent(serviceKey,
                        new myStatItem(serviceKey, rate, interval));
                statItem = stats.get(serviceKey);
            }
            //这里的isAllowable()主要是执行rate值减一操作，如果当前rate值为0，则拒绝本次服务调用
            return statItem.isAllowable();
        } else {
            //小于等于0表示不限速，所以需要remove掉限制
            myStatItem statItem = stats.get(serviceKey);
            if (statItem != null) {
                stats.remove(serviceKey);
            }
        }

        return true;
    }
}
